package tasks;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.StreamTokenizer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import system.MapperResult;
import system.ReducerResult;

import api.TaskBase;

/**
 *   MapReduce implementation of Word Count task
 *   
 *     @author Mayuri Karra, Bing Wei
 *    
 *   
 */

public class WordCountTask extends TaskBase{

	private static final long serialVersionUID = 1L;

	public WordCountTask(String inputFile, String outputFile) {
		super(inputFile, outputFile);
	}

	/**
	 * Map method for word count index. It takes a <key, value> pair
	 * and returns a list of MapperResult pairs
	 * @param key that is to mapped
	 * @param value 
	 * @return a list of MapperResults which are essentially <key,value> pairs 
	 */
	@Override
	public List<MapperResult> map(Object key, Object value) {
		// TODO Auto-generated method stub
		BufferedReader input = null;
	    List<MapperResult> keyValueList = new ArrayList<MapperResult>();

		try {
			input =  new BufferedReader(new FileReader((String)key));
			//System.out.println("reading: " + key);
	        Map<String, Integer> hashMap = new HashMap <String, Integer>();
	
			try {
				StreamTokenizer st =  new StreamTokenizer(new FileReader((String)key));
				//System.out.println("reading: " + key);
		
		        while (st.nextToken() != StreamTokenizer.TT_EOF){
		        	if(st.ttype == StreamTokenizer.TT_WORD){
		        		String word = st.sval;		  		
		        		if (hashMap.containsKey(word)){
		         			hashMap.put(word,hashMap.get(word) + 1);
		         		}else{
		         			hashMap.put(word, 1);
		         		}	     	        	      	         		
		        	}
		        }
			}
		    catch (IOException ex){
		      ex.printStackTrace();
		    }
		  
	        for(Map.Entry<String, Integer> entry: hashMap.entrySet()){
	        	keyValueList.add(new MapperResult(entry.getKey(), entry.getValue()));
	        }
        	input.close();
        }
	    catch (IOException ex){
	      ex.printStackTrace();
	    }
	    finally {
	        try {
				input.close();
			} catch (IOException e) {
				e.printStackTrace();
			}	        
	    }
		
      return keyValueList;
		
	}


	/**
	 * Method for the reduce operation of word count index task. 
	 * It takes a list of <key, list<value>> pairs
	 * and returns ReducerResult
	 * 
	 * @param key
	 * @param values
	 * @return ReducerResult which has <key, value>
	 */

	@Override
	public  ReducerResult reduce(Object key, List<Object> values) {
		// TODO Auto-generated method stub
		int sum = 0;
	    for(Object aValue:values){
	    	sum += Integer.parseInt((String)aValue);
	    }
	    return new ReducerResult(key, new Integer(sum));
		
	}
	
}